package com.match.controller;

import java.math.BigDecimal;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.IMap;
import com.jfinal.kit.PropKit;
import com.match.common.model.dto.ClusterInfo;
import com.match.common.model.dto.Depth;
import com.match.plugin.hazelcast.HazelcastKit;
import com.match.plugin.hazelcast.HazelcastPlugin;
import com.match.plugin.kafka.Kafka;

public class PushDepthTask implements Runnable {
	
	HazelcastInstance hzInstance = HazelcastPlugin.hazelcastInstance;
	
	//这里的交易队，可选择配置在缓存中，实现动态添加交易队
	static List<String> coinTeams = Arrays.asList(PropKit.get("coinTeams").split(","));
	

	/**
	 * @Title: PushOrder
	 * @Description: TODO(推送深度)
	 */
	public void run() {
		try {
			if (!isMasterNode()) {
				return;
			}
			if (!coinTeams.isEmpty()) {
				for (String coinTeam : coinTeams) {
					//  买盘
					List<Depth> buyList = this.getMarketDepth(coinTeam, Boolean.TRUE);
					//  卖盘
					List<Depth> sellList = this.getMarketDepth(coinTeam, Boolean.FALSE);
					// 发送数据处理
					Map<String, List<Depth>> map = new HashMap<String, List<Depth>>();
					map.put("buy", buyList);
					map.put("sell", sellList);
					// 推送深度
					Kafka.send(Kafka.tradeDepth, Kafka.tradeDepth, new Date().getTime(),map);
				}
			}
		} catch (Exception e) {
			System.out.println("深度数据处理错误：" + e);
			e.printStackTrace();
		}
	}

	/**
	 * 获取行情深度
	 * 
	 * @param coinTeam 交易队
	 * @param isBuy    是否是买
	 * @return List<Depth>
	 */
	public List<Depth> getMarketDepth(String coinTeam, Boolean isBuy) {
		// XBIT-USDT 买盘
		IMap<BigDecimal, BigDecimal> buyMap = hzInstance.getMap(HazelcastKit.getMatchKey(coinTeam, isBuy));
		List<Depth> depths = new ArrayList<Depth>();
		if (buyMap.size() > 0) {
			List<Depth> list = new ArrayList<Depth>();
			if (isBuy) {
				list = buyMap.entrySet().stream().sorted(Entry.<BigDecimal, BigDecimal>comparingByKey().reversed())
						.map(obj -> 
							new Depth(obj.getKey().toString(), obj.getValue().toString(),obj.getValue().toString(), 1, coinTeam, isBuy)
						).limit(100)
						.collect(Collectors.toList());
			} else {
				list = buyMap.entrySet().stream().sorted(Entry.<BigDecimal, BigDecimal>comparingByKey())
						.map(obj -> 
							new Depth(obj.getKey().toString(), obj.getValue().toString(),
								obj.getValue().toString(), 1, coinTeam, isBuy)
						).limit(100)
						.collect(Collectors.toList());
			}
			list.stream().reduce(new Depth("0", "0", "0", 1, coinTeam, isBuy), (one, two) -> {
				one.setTotal((new BigDecimal(one.getTotal()).add(new BigDecimal(two.getNumber()))).toString());
				depths.add(
						new Depth(two.getPrice(), two.getNumber(), one.getTotal(), two.getPlatform(),
						two.getCoinTeam(), two.getIsBuy())
				);
				return one;
			});
		} else {
			Depth depth = new Depth("0", "0", "0", 1, coinTeam, isBuy);
			depths.add(depth);
		}
		return depths;
	}
	
	/**
	 * @Title: isMasterNode
	 * @Description: TODO(主节点判断)
	 * @param  参数
	 * @return void 返回类型
	 * @throws
	 */
	private Boolean isMasterNode() {
		try {
			IMap<String, ClusterInfo> map = hzInstance.getMap("Cluster-IP");
			long current = System.currentTimeMillis();
			String ip = InetAddress.getLocalHost().getHostAddress();//获得本机IP  
			if (!map.containsKey(ip)) {
				map.put(ip, new ClusterInfo(ip, current, current));
			}
			Comparator<ClusterInfo> comparing = Comparator.comparing(ClusterInfo::getStartTime);
			ClusterInfo ci =map.values().stream().min(comparing).get();
			if (ci.getIp().equals(ip)) {
				ci.setUpdateTime(System.currentTimeMillis());
				map.put(ip, ci);
				return Boolean.TRUE;
			}else {
				long interval = current - ci.getUpdateTime();
				if (ci.getStartTime()!=ci.getUpdateTime()&&interval>3000) {
					map.delete(ci.getIp());
				}
			}
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		return Boolean.FALSE;
	}


}
